package com.zidongxiangxi.clock.server.business;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.zidongxiangxi.clock.http.constants.ClockErrorCodeConstant;
import com.zidongxiangxi.clock.http.constants.RabbitConstant;
import com.zidongxiangxi.clock.http.exception.ClockException;
import com.zidongxiangxi.clock.server.config.ClockConfig;
import com.zidongxiangxi.clock.server.constants.RedisConstant;
import com.zidongxiangxi.clock.server.manager.ClockTaskManger;
import com.zidongxiangxi.clock.server.repository.ClockDao;
import com.zidongxiangxi.clock.server.repository.entity.Clock;
import com.zidongxiangxi.clock.server.utils.ClockUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

/**
 * This is Description
 *
 * @author chenxudong
 * @date 2021/12/27
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class ClockBusiness {
    private static final int VALID_CLOCKS = 1;
    private static final int INVALID_CLOCKS = 2;

    private final ClockConfig config;
    private final ThreadPoolExecutor asyncExecutor;

    private final ClockDao clockManger;
    private final ClockTaskManger clockTaskManger;

    private final TransactionTemplate transactionTemplate;
    private final StringRedisTemplate redisTemplate;
    private final RabbitTemplate rabbitTemplate;

    public Long createClock(String application, String businessId, String payload, Long targetRingAt,
        Long latestRingAt) throws ClockException {
        if (!StringUtils.isEmpty(application) || !StringUtils.isEmpty(businessId)) {
            if (StringUtils.isEmpty(application)) {
                throw ClockException.newInstance(ClockErrorCodeConstant.APPLICATION_MUST_NOT_EMPTY);
            }
            if (StringUtils.isEmpty(businessId)) {
                throw ClockException.newInstance(ClockErrorCodeConstant.BUSINESS_ID_MUST_NOT_EMPTY);
            }
        }
        if (StringUtils.isEmpty(payload)) {
            if (StringUtils.isEmpty(businessId)) {
                throw ClockException.newInstance(ClockErrorCodeConstant.BUSINESS_ID_MUST_NOT_EMPTY);
            }
        }
        if (Objects.isNull(targetRingAt)) {
            throw ClockException.newInstance(ClockErrorCodeConstant.TARGET_RING_TIME_MUST_NOT_EMPTY);

        }
        long current = System.currentTimeMillis();
        if (targetRingAt < current) {
            throw ClockException.newInstance(ClockErrorCodeConstant.TARGET_RING_TIME_BEFORE_NOW);
        }
        if (Objects.nonNull(latestRingAt) && latestRingAt < targetRingAt) {
            throw ClockException.newInstance(ClockErrorCodeConstant.LATEST_RING_TIME_BEFORE_TARGET_RING_TIME);
        }

        Clock clock = new Clock();
        clock.setApplication(application);
        clock.setBusinessId(businessId);
        clock.setPayload(payload);
        clock.setTargetRingAt(targetRingAt);
        clock.setLatestRingAt(latestRingAt);
        clock.setNextTryTime(targetRingAt);
        clock.setRetryTimes(0);
        clock.setCreateTime(new Date(current));
        clock.setUpdateTime(clock.getCreateTime());

        return createClock(clock);
    }

    private Long createClock(Clock clock) {
        try {
            transactionTemplate.execute(status -> {
                clockManger.save(clock);
                if (inPreLoadPeriod(clock.getNextTryTime())) {
                    Long newNextTryTime = ClockUtils.getNewNextTryTime();
                    clockManger.changeNextTryTime(clock.getId(), newNextTryTime);
                    clockTaskManger.addClockTask(clock);
                }
                return null;
            });
            return clock.getId();
        } catch (DuplicateKeyException e) {
            Clock existClock = clockManger.getAlarmClockByApplicationAndBusinessId(clock.getApplication(), clock.getBusinessId());
            return existClock.getId();
        } catch (Exception e) {
            log.error("create clock fail , clock = {} ", JSON.toJSONString(clock));
            throw ClockException.newInstance(ClockErrorCodeConstant.ADD_CLOCK_FAIL);
        }
    }

    public void deleteClock(String application, String businessId) {
        Clock existClock = clockManger.getAlarmClockByApplicationAndBusinessId(application, businessId);
        if (Objects.isNull(existClock)) {
            return;
        }
        deleteClock(existClock);
    }

    public void deleteClockById(Long clockId) {
        Clock existClock = clockManger.getById(clockId);
        if (Objects.isNull(existClock)) {
            return;
        }
        deleteClock(existClock);
    }

    private void deleteClock(Clock clock) {
        if (inPreLoadPeriod(clock.getNextTryTime())) {
            redisTemplate.opsForSet().add(RedisConstant.DELETE_CLOCK_KEY, String.valueOf(clock.getId()));
        } else {
            clockManger.removeById(clock.getId());
        }
    }

    private boolean inPreLoadPeriod(Long nextTryTime) {
        return nextTryTime - System.currentTimeMillis() < config.getDatabase().getPreLoadMilliseconds();
    }

    public void ringClockAsync(List<Clock> clocks, long batchBeginTime) {
        CompletableFuture.runAsync(() -> ringClock(clocks, batchBeginTime), asyncExecutor);
    }

    public void ringClock(List<Clock> clocks, long batchBeginTime) {
        long now = System.currentTimeMillis();

        Set<String> keySets =
            clocks.stream().map(clock -> clock.getId().toString()).collect(Collectors.toSet());
        Set<String> deleteIdStrSet = redisTemplate.opsForSet().intersect(RedisConstant.DELETE_CLOCK_KEY, keySets);
        if (CollectionUtils.isNotEmpty(deleteIdStrSet)) {
            List<Long> deletedIds = deleteIdStrSet.stream().map(Long::valueOf).collect(Collectors.toList());
            // 先删mysql，再删redis，防止数据一致性问题
            clockManger.removeByIds(deletedIds);
            redisTemplate.opsForSet().remove(RedisConstant.DELETE_CLOCK_KEY, deleteIdStrSet);
        }

        Map<Integer, List<Clock>> alarmClocksMap =
            clocks.stream()
                .collect(Collectors.groupingBy(alarmClock -> {
                    if (Objects.nonNull(alarmClock.getLatestRingAt()) && alarmClock.getLatestRingAt() < now) {
                        return INVALID_CLOCKS;
                    }
                    return VALID_CLOCKS;
                }));
        List<Clock> validClocks = alarmClocksMap.get(VALID_CLOCKS);
        List<Clock> invalidClocks = alarmClocksMap.get(INVALID_CLOCKS);

        if (CollectionUtils.isNotEmpty(invalidClocks)) {
            List<Long> ids = invalidClocks.stream().map(Clock::getId).collect(Collectors.toList());
            clockManger.removeByIds(ids);
            log.debug("clock[{}] latest ring at is less than now[{}] , ignore clock !", ids, now);
        }

        if (CollectionUtils.isEmpty(validClocks)) {
            return;
        }
        validClocks.forEach(this::ringClock);
        log.info("batch job begin at {} , end at {}, total job is {}", batchBeginTime,
            System.currentTimeMillis() / 1000, clocks.size());
    }

    public void ringClock(Clock clock) {
        String payload = clock.getPayload();
        byte[] bytes = payload.getBytes(StandardCharsets.UTF_8);
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
        messageProperties.setContentEncoding(StandardCharsets.UTF_8.name());
        messageProperties.setContentLength(bytes.length);
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        Message message = new Message(payload.getBytes(StandardCharsets.UTF_8), messageProperties);

        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(clock.getId().toString());

        rabbitTemplate.send(RabbitConstant.CLOCK_CALLBACK_EXCHANGE, clock.getApplication(), message, correlationData);
    }
}
